package com.zzc.legou.log.aop.utils;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.concurrent.LinkedBlockingDeque;

@Component
public class LogContainer {
    //异步线程池的使用
    private LogThread logThread;

    @Value("${kafka.topic}")
    private String topic;

    //使用阻塞线程
    private LinkedBlockingDeque<String> logs = new LinkedBlockingDeque<>();

    public LogContainer() {
        this.logThread = new LogThread();
        this.logThread.start();
    }

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 存入一条消息 并将该消息 发送给kafka进行消费处理
     *
     * @param log 传输的消息信息
     */
    public void putLog(String log) {
        logs.offer(log);
    }


    /**
     * 异步日志线程 实时从队列中获取内容
     */
    class LogThread extends Thread {

        @Override
        public void run() {
            while (true) {
                /**
                 * 代码的优化
                 * 当前线程批量获取多条日志消息 投递kafka   批量
                 *
                 */
                String msg = logs.poll();
                if (!StringUtils.isEmpty(msg)) {
                    kafkaTemplate.send(topic, msg);
                }
            }
        }
    }
}
